﻿using SimpleSockets;
using System;
using System.Buffers;
using System.Collections.Generic;
using System.IO.Pipelines;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;

namespace PipeSockets
{
    public abstract class AbstractConnection : IDuplexPipe
    {
        protected readonly PipeOptions SendPipeOptions;
        protected readonly PipeOptions ReceivePipeOptions;

        protected readonly Pipe _sendPipe;
        protected readonly Pipe _receivePipe;

        protected EndPoint LocalEndPoint { get; }
        protected EndPoint RemoteEndPoint { get; }

        private readonly IProtocolParser _protocolParser;
        public Action<AbstractConnection, ReadOnlySequence<byte>> OnMessageReceived;

        private volatile bool _isStopping;

        protected AbstractConnection(IProtocolParser protocolParser, PipeOptions sendPipeOptions, PipeOptions receivePipeOptions, EndPoint localEndPoint, EndPoint remoteEndPoint)
        {
            SendPipeOptions = sendPipeOptions ?? PipeOptions.Default;
            ReceivePipeOptions = receivePipeOptions ?? PipeOptions.Default;

            _sendPipe = new Pipe(SendPipeOptions);
            _receivePipe = new Pipe(ReceivePipeOptions);

            _protocolParser = protocolParser;

            LocalEndPoint = localEndPoint;
            RemoteEndPoint = remoteEndPoint;
        }

        public PipeReader Input => _receivePipe.Reader;
        public PipeWriter Output => _sendPipe.Writer;
        private Task sendTask;
        private Task receiveTask;
        private Task processTask;
        protected abstract ValueTask<int> ReceiveAsync(Memory<byte> buffer, CancellationToken cancellationToken = default);
        protected abstract ValueTask<int> SendAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default);

        public async Task StartReceivingAsync(CancellationToken cancellationToken = default)
        {
            while (true)
            {
                try
                {
                    var buffer = _receivePipe.Writer.GetMemory(1);
                    int bytesReceived = await ReceiveAsync(buffer, cancellationToken).ConfigureAwait(false);
                    if (bytesReceived == 0)
                    {
                        break;
                    }
                    _receivePipe.Writer.Advance(bytesReceived);
                }
                catch (Exception ex)
                {
                    // 记录异常信息，考虑是否需要关闭连接或采取其他恢复措施
                    break;
                }

                await _receivePipe.Writer.FlushAsync(cancellationToken).ConfigureAwait(false);
            }

            await _receivePipe.Writer.CompleteAsync().ConfigureAwait(false);
        }

        public async Task StartSendingAsync(CancellationToken cancellationToken = default)
        {
            while (true)
            {
                var readResult = await _sendPipe.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
                var buffer = readResult.Buffer;

                try
                {
                    foreach (var segment in buffer)
                    {
                        await SendAsync(segment, cancellationToken).ConfigureAwait(false);
                    }

                    if (readResult.IsCompleted)
                    {
                        break;
                    }
                }
                catch (Exception ex)
                {
                    // 记录异常信息，考虑是否需要关闭连接或采取其他恢复措施
                    break;
                }
                finally
                {
                    _sendPipe.Reader.AdvanceTo(buffer.Start, buffer.End);
                }
            }

            await _sendPipe.Reader.CompleteAsync().ConfigureAwait(false);
        }
        private async Task ProcessReceivedDataAsync(CancellationToken cancellationToken = default)
        {
            while (true)
            {
                var readResult = await Input.ReadAsync(cancellationToken).ConfigureAwait(false);
                var buffer = readResult.Buffer;
                try
                {
                    while (TryReadMessage(ref buffer, out ReadOnlySequence<byte> message))
                    {
                        OnMessageReceived(this, message);
                    }

                    if (readResult.IsCompleted)
                    {
                        break;
                    }
                }
                catch (Exception ex)
                {
                    // 记录异常信息，考虑是否需要关闭连接或采取其他恢复措施
                    break;
                }
                finally
                {
                    // 成功处理一个消息后，更新Input的当前位置，记录已消费的消息长度
                    Input.AdvanceTo(buffer.Start, buffer.End);
                }
            }
        }

        private bool TryReadMessage(ref ReadOnlySequence<byte> buffer, out ReadOnlySequence<byte> message)
        {
            if (_protocolParser.Decode(buffer, out message, out long consumedBytes))
            {
                // 更新buffer引用，使其指向剩余未处理的数据
                buffer = buffer.Slice(consumedBytes);
                return true;
            }
            return false;
        }


        public async Task SendFrameAsync(byte[] message, CancellationToken cancellationToken = default)
        {
            var data = _protocolParser.Encode(message);
            Output.Write(data.Span);
            await Output.FlushAsync(cancellationToken).ConfigureAwait(false);
        }
        public Task StartAsync(CancellationToken cancellationToken = default)
        {
            receiveTask = StartReceivingAsync(cancellationToken);
            sendTask = StartSendingAsync(cancellationToken);
            processTask = ProcessReceivedDataAsync(cancellationToken);
            return Task.CompletedTask;
        }

        public async Task StopAsync(CancellationToken cancellationToken = default)
        {
            await _sendPipe.Writer.CompleteAsync().ConfigureAwait(false);
            await _receivePipe.Reader.CompleteAsync().ConfigureAwait(false);
            await OnStopAsync(cancellationToken).ConfigureAwait(false);
        }

        protected virtual Task OnStopAsync(CancellationToken cancellationToken = default)
        {
            return Task.CompletedTask;
        }
    }
}
